import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author 帅小伙
 * @date 2022/1/3
 * @description   3种发布确认策略
 */
public class PublicConfirmsTest {

    public static int MESSAGE_COUNT = 100000;


    public static void main(String[] args) throws Exception {
       // publishMessageIndividual();
        publishMessageBatch();
        publishMessageAsync();
    }


    // 单个消息确认发布
    //检测 发送:100条消息耗时:1958
    //不检测 发送:100条消息耗时:16
    public static void publishMessageIndividual() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString().substring(1, 5);
        channel.queueDeclare(queueName, false, false, false, null);
        // 开启发布确认   发布消息更加安全
        channel.confirmSelect();

        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", queueName, null, message.getBytes());
            // 等待broker接收到生产者的消息
            boolean flag = false;
            do {
                flag = channel.waitForConfirms();
            } while (!flag);
        }
        long end = System.currentTimeMillis();
        System.out.println("发送:" + MESSAGE_COUNT + "条消息耗时:" + (end - begin));
    }




    public static void publishMessageBatch() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString().substring(1, 5);
        channel.queueDeclare(queueName, false, false, false, null);
        channel.confirmSelect();

        // 批量大小
        //多少个消息还未确认的消息
        int batchSize = 50;
        int outstandingMessageCount = 0;

        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", queueName, null, message.getBytes());
            // 等待broker接收到生产者的消息
            if(outstandingMessageCount == batchSize){
                boolean flag = false;
                do {
                    flag = channel.waitForConfirms();
                } while (!flag);
                outstandingMessageCount = 0;
            }
        }
        if(outstandingMessageCount > batchSize){
            boolean flag = false;
            do {
                flag = channel.waitForConfirms();
            } while (!flag);
        }
        long end = System.currentTimeMillis();
        System.out.println("发送:" + MESSAGE_COUNT + "条消息批量确认耗时:" + (end - begin));
    }





    /*
    *  异步消息发布确认
    *   数据越多 异步越快
    * */
    public static void publishMessageAsync() throws Exception {
        long begin = System.currentTimeMillis();
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString().substring(1, 5);
        channel.queueDeclare(queueName, false, false, false, null);
        // 开启发布确认   发布消息更加安全
        channel.confirmSelect();

        /*
        * 签收小于等于    所以不能用hashmap  使用 ConcurrentSkipListMap
        * 1.线程安全
        * 2.将序号与消息进行关联
        * 3.只要给定序号  就会将<=当前序号的值作为一个map返回
        * */
        ConcurrentSkipListMap<Long, String> outstandingComfirms = new ConcurrentSkipListMap();


        /*
        * 确认收到消息的回调
        * */
        ConfirmCallback ackCallback = (sequenceNumber,multiple)->{

            if(multiple){
                //签收小于等于 sequenceNumber 的消息
                // 取出<= sequenceNumber 的map集合
                ConcurrentNavigableMap<Long, String> map = outstandingComfirms.headMap(sequenceNumber);
                //删除已经确定收到的一部分map
                map.clear();
            }else{
                outstandingComfirms.remove(sequenceNumber);
            }

        };
        ConfirmCallback nackCallback = (sequenceNumber,multiple)->{
            if(multiple){
                //签收小于等于 sequenceNumber 的消息
                ConcurrentNavigableMap<Long, String> map = outstandingComfirms.headMap(sequenceNumber);
            }else{
                channel.basicPublish("", queueName, null, outstandingComfirms.get(sequenceNumber).getBytes());
            }

        };

        /*
         * 收到消息的回调   未收到消息的回调
         * 两个回调函数
         * ConfirmCallback ackCallback,     void handle(long deliveryTag, boolean multiple) throws IOException;
         *  ConfirmCallback nackCallback     void handle(long deliveryTag, boolean multiple) throws IOException;
         * */

        channel.addConfirmListener(ackCallback,nackCallback);


        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            // 获取消息的下一个标号    1-
            outstandingComfirms.put(channel.getNextPublishSeqNo(),message);
            channel.basicPublish("", queueName, null, message.getBytes());
        }
        long end = System.currentTimeMillis();
        System.out.println("发送:" + MESSAGE_COUNT + "条消息异步处理耗时:" + (end - begin));
    }



}
